package org.pupil.rpc.consumer;

import io.netty.channel.DefaultEventLoop;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;
import org.pupil.rpc.common.RpcFuture;
import org.pupil.rpc.common.RpcRequestHolder;
import org.pupil.rpc.common.RpcServiceNameBuilder;
import org.pupil.rpc.common.constants.MsgType;
import org.pupil.rpc.common.constants.ProtocolConstants;
import org.pupil.rpc.config.RpcProperties;
import org.pupil.rpc.filter.FilterConfig;
import org.pupil.rpc.filter.FilterData;
import org.pupil.rpc.protocol.MsgHeader;
import org.pupil.rpc.protocol.RpcProtocol;
import org.pupil.rpc.protocol.RpcRequest;
import org.pupil.rpc.protocol.RpcResponse;
import org.pupil.rpc.router.LoadBalancer;
import org.pupil.rpc.router.LoadBalancerFactory;
import org.pupil.rpc.router.ServiceMetaRes;
import org.pupil.rpc.tolerant.FaultTolerantFactory;
import org.pupil.rpc.tolerant.FaultTolerantStrategy;
import org.pupil.rpc.tolerant.FaultTolerantStrategyShareContainer;
import org.springframework.util.ObjectUtils;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;


/**
 * 服务代理
 */
@Slf4j
public class RpcInvokerProxy implements InvocationHandler {

    //服务版本
    private String serviceVersion;
    //超时时间
    private long timeout;
    //负载平衡器类型
    private String loadBalancerType;
    //容错机制类型
    private String faultTolerantType;
    //重试次数
    private long retryCount;

    public RpcInvokerProxy(String serviceVersion, long timeout, String loadBalancerType, String faultTolerantType, long retryCount) {
        this.serviceVersion = serviceVersion;
        this.timeout = timeout;
        this.loadBalancerType = loadBalancerType;
        this.faultTolerantType = faultTolerantType;
        this.retryCount = retryCount;
    }

    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        RpcProtocol<RpcRequest> protocol = new RpcProtocol<>();

        // 构建消息头
        MsgHeader header = new MsgHeader();
        header.setMagic(ProtocolConstants.MAGIC);
        header.setVersion(ProtocolConstants.VERSION);
        long requestId = RpcRequestHolder.REQUEST_ID_GEN.incrementAndGet(); //获取请求id
        header.setRequestId(requestId);
        // 序列化规则字节数组
        final byte[] serialization = RpcProperties.getInstance().getSerialization().getBytes();
        header.setSerializationLen(serialization.length);
        header.setSerializations(serialization);
        // 压缩化规则字节数组
        final  byte[] compression = RpcProperties.getInstance().getCompression().getBytes();
        header.setCompressLen(compression.length);
        header.setCompression(compression);
        // 设置消息类型
        header.setMsgType((byte) MsgType.REQUEST.ordinal());
        header.setStatus(ProtocolConstants.status); //消息状态未知
        // 设置消息头
        protocol.setHeader(header);

        // 设置请求体
        RpcRequest request = new RpcRequest();
        request.setServiceVersion(this.serviceVersion);
        request.setClassName(method.getDeclaringClass().getName());
        request.setMethodName(method.getName());
        request.setParameterTypes(method.getParameterTypes());
        request.setData(ObjectUtils.isEmpty(args) ? new Object[0] : args);
        request.setDataClass(ObjectUtils.isEmpty(args) ? null : args[0].getClass());
        request.setServiceAttachments(RpcProperties.getInstance().getServiceAttachments());
        request.setClientAttachments(RpcProperties.getInstance().getClientAttachments());

        //拦截器的上下文
        final FilterData filterData = new FilterData(request);
        try {
            // 调用消费端前置拦截器的doFilter，进行处理
            FilterConfig.getClientBeforeFilterChain().doFilter(filterData);
        } catch (Throwable e) {
            throw e;
        }

        // 设置消息体
        protocol.setBody(request);

        // 创建客户端
        RpcConsumer rpcConsumer = new RpcConsumer();

        // 创建服务名
        String serviceName = RpcServiceNameBuilder.buildServiceKey(request.getClassName(), request.getServiceVersion());
        // 获取参数数组
        Object[] params = new Object[]{request.getData()};

        // 获取负载均衡策略
        final LoadBalancer loadBalancer = LoadBalancerFactory.get(loadBalancerType);

        // 根据负载均衡策略获取对应服务
        final ServiceMetaRes serviceMetaRes = loadBalancer.select(params, serviceName);


        // 获取容错策略数据共享容器
        FaultTolerantStrategyShareContainer shareContainer = FaultTolerantStrategyShareContainer.getInstance();
        // 设置运行次数
        shareContainer.setCount(1);
        // 设置当前服务节点
        shareContainer.setCurServiceMeta(serviceMetaRes.getCurServiceMeta());
        // 设置剩余当前服务节点（空闲状态）
        shareContainer.setOtherServiceMeta(serviceMetaRes.getOtherServiceMeta());
        // 设置服务名
        shareContainer.setServiceName(serviceName);

        // 最大重试次数
        long retryCount = this.retryCount;
        RpcResponse response = null;
        // 重试机制
        while (shareContainer.getCount() <= retryCount) {
            // 创建异步处理返回数据通道
            RpcFuture<RpcResponse> future = new RpcFuture<>(new DefaultPromise<>(new DefaultEventLoop()), timeout);
            // 绑定请求通道
            RpcRequestHolder.REQUEST_MAP.put(requestId, future);
            try {
                // 发送消息
                rpcConsumer.sendRequest(protocol, shareContainer.getCurServiceMeta());
                // 等待响应数据返回，设置最大等待时间，当超时时报超时错误
                response = future.getPromise().get(future.getTimeout(), TimeUnit.MILLISECONDS);

                // 如果有异常并且没有其他服务
                if (response.getException() != null && shareContainer.getOtherServiceMeta().size() == 0) {
                    throw response.getException();
                }
                if (response.getException() != null) {
                    throw response.getException();
                }

                log.info("rpc 调用成功，serviceName：{}", serviceName);
                try {
                    // 客户端后处理器添加上下文数据
                    FilterConfig.getClientAfterFilterChain().doFilter(filterData);
                } catch (Throwable e) {
                    throw e;
                }

                // 返回数据
                return response.getData();
            } catch (Throwable e) {
                // 错误信息
                String errorMsg = e.toString();
                // 设置错误信息
                shareContainer.setErrorMsg(errorMsg);
                // 设置响应消息
                shareContainer.setRpcResponse(response);
                // 获取容错策略
                FaultTolerantStrategy faultTolerantStrategy = FaultTolerantFactory.get(faultTolerantType);
                // 进行容错处理
                faultTolerantStrategy.handler();
                // 判断是否有返回值
                if (shareContainer.getIsReturnValue()) {
                    return shareContainer.getReturnValue();
                }
            }
        }

        throw new RuntimeException("rpc 调用失败，超过最大重试次数: {}" + retryCount);
    }
}
